package kk

import (
	"context"
	"fmt"
	"github.com/segmentio/kafka-go"
	"log"
	"time"
)

// KafkaReader 消息读出者  消费者
type KafkaReader struct {
	R *kafka.Reader
}

func (r *KafkaReader) readMeg() {
	for {
		m, err := r.R.ReadMessage(context.Background())
		if err != nil {
			log.Printf("kafka send readmessage err %s", err.Error())
			continue
		}
		fmt.Printf("message at topic/partition/offset %v/%v/%v: %s = %s\n", m.Topic, m.Partition, m.Offset, string(m.Key), string(m.Value))
	}
}
func (r *KafkaReader) Close() {
	r.R.Close()
}

// GetReader 根据应用程序的需求和性能需求来决定需要多少个Kafka生产者和消费者
// 而Kafka Brokers的数量通常由集群的规模和负载要求来决定。
func GetReader(brokers []string, groupId, topic string) *KafkaReader {
	//创建一个消费者 在处理大量数据或需要快速处理数据的场景下，可能需要多个消费者来并行读取数据。
	r := kafka.NewReader(kafka.ReaderConfig{
		//kafka的服务端口 一个Kafka消费者可以连接到一个或多个Kafka Brokers
		Brokers:  brokers,
		GroupID:  groupId, //同一个组下的consumer  协同工作  共同消费topic队列中的内容
		Topic:    topic,
		MaxBytes: 10e6, // 10MB
		//上报每次读到数据的位置
		CommitInterval: time.Second, // flushes commits to Kafka every second
	})
	k := &KafkaReader{R: r}
	//不断地去读消息
	//go k.readMeg()
	return k
}
